package com.fannuo.demo.Api;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fannuo.demo.Entity.Journal;
import com.fannuo.demo.Util.FileUtil;
import com.fannuo.demo.Util.hbase.HbaseClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Component
@Slf4j
public class HbaseApi {
    private HbaseClient hbaseClient;
    public HbaseApi() {
        hbaseClient = new HbaseClient();
    }

    public HbaseClient getClient() {
        return hbaseClient;
    }

    // 导入元数据文件
    public void put(String path) {
        log.info("开始导入 "+path);
        try {
            // 获取连接对象
            Connection connection = hbaseClient.getConnection();
            // 读取json文件
            List<JSONObject> list = FileUtil.getJSONList(new File(path));
            // 表名
            String tableName = "test_journal";
            for(JSONObject one : list) {
                // 获取字段
                String rowKey = one.getString("id");
                String title = one.getString("title");
                String authors = one.getString("authors");
                // 将一条记录写入到hbase中
                Table table = connection.getTable(TableName.valueOf(tableName));
                Put put = new Put(rowKey.getBytes());
                put.addColumn("Meta".getBytes(), "title".getBytes(), title.getBytes());
                put.addColumn("Meta".getBytes(), "authors".getBytes(), authors.getBytes());
                table.put(put);
                table.close();

                log.info("已导入 " + rowKey);
            }
            log.info("导入完成 "+ path);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 获取hbase记录
    public List<Journal> getList(int limit) {
        List<Journal> journalList = new ArrayList<>();
        try {
            Connection connection = hbaseClient.getConnection();
            Table table = connection.getTable(TableName.valueOf("test_journal"));
            Scan scan = new Scan();
            // 查找标题为 蒲松龄之贫困书写刍议 的数据
            SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes("Meta"),Bytes.toBytes("title"),
                    CompareOperator.EQUAL,Bytes.toBytes("蒲松龄之贫困书写刍议"));
            filter.setFilterIfMissing(true);
            // 添加筛选条件，如需筛选可取消注释
//            scan.setFilter(filter);
            // 设置最大记录数
            scan.setLimit(limit);
            ResultScanner rs = table.getScanner(scan);
            for(Result result : rs){
                String rowKey = Bytes.toString(result.getRow());
                String title = Bytes.toString(result.getValue("Meta".getBytes(), "title".getBytes()));
                String authors = Bytes.toString(result.getValue("Meta".getBytes(), "authors".getBytes()));
                // 将作者解析为数组
                List<String> authorList = JSON.parseArray(authors, String.class);

                Journal journal = new Journal();
                journal.setId(rowKey);
                journal.setTitle(title);
                journal.setAuthors(authorList);
                journalList.add(journal);

                log.info("hbase:获取到数据：{}", JSON.toJSONString(journal));
            }
            table.close();
            return journalList;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    // 根据rowKey获取一条记录
    public Journal getOne(String rowKey) {
        try {
            Connection connection = hbaseClient.getConnection();
            Table table = connection.getTable(TableName.valueOf("test_journal"));
            Get get = new Get(rowKey.getBytes());
            Result result = table.get(get);

            // 解析数据
            String title = Bytes.toString(result.getValue("Meta".getBytes(), "title".getBytes()));
            String authors = Bytes.toString(result.getValue("Meta".getBytes(), "authors".getBytes()));
            // 将作者解析为数组
            List<String> authorList = JSON.parseArray(authors, String.class);

            Journal journal = new Journal();
            journal.setId(rowKey);
            journal.setTitle(title);
            journal.setAuthors(authorList);

            log.info("hbase:获取到数据：{}", JSON.toJSONString(journal));
            table.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public void close() {
        try {
            hbaseClient.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
